package com.heima.kafka.sample;


import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * 生产者
 */
public class ProducerQuickStart {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //1.kafka连接配置信息
        Properties properties = new Properties();
        //kafka的连接地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.136.152:9092");
        //key和value的序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        //ack配置，消息确认机制
        properties.put(ProducerConfig.ACKS_CONFIG,"all");
        //设置重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG,10);
        //消息压缩
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"lz4");
        //2.创建生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);
        //3.发送消息
        /**
         * 第一个参数：topic
         * 第二个参数：消息的key
         * 第三个参数：消息的value
         */
        System.out.println("正在发送消息");
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String,String> kvProducerRecord = new ProducerRecord<String,String>("itcast-topic-input","hello kafka");
            producer.send(kvProducerRecord);
        }
        System.out.println("发送消息成功");

        //同步发送消息
        //RecordMetadata recordMetadata = producer.send(kvProducerRecord).get();
        //System.out.println(recordMetadata.offset());//获取偏移量
        //异步发送消息
        //producer.send(kvProducerRecord, new Callback() {
        //    @Override
        //    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        //        if (e != null){
        //            System.out.println("记录异常信息到日志表中");
        //        }
        //        System.out.println(recordMetadata.offset());
        //    }
        //});
        //
        //System.out.println("消息发送成功");
        //4.关闭消息通道 必须要关闭，否则消息发送不成功
        producer.close();
    }
}